package com.zili.sink;

import com.zili.source.Event;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author : ranzlupup
 * @since : 2023/3/8 15:14
 */
public class SinkToHBase {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        DataStreamSource<Event> stream = env.fromElements(
                new Event("Mary1", "./home", 1000L),
                new Event("Bob2", "./cart", 2000L),
                new Event("Alice3", "./prod?id=1", 3000L),
                new Event("Cary4", "./home", 4000L),
                new Event("Mary5", "./home", 1000L),
                new Event("Bob6", "./cart", 2000L),
                new Event("Alice7", "./prod?id=1", 3000L),
                new Event("Cary8", "./home", 4000L),
                new Event("Mary9", "./home", 1000L),
                new Event("Bob2", "./cart", 2000L),
                new Event("Alice3", "./prod?id=1", 3000L),
                new Event("Cary4", "./home", 4000L)
        );
        stream.addSink(
                JdbcSink.sink(
                        "INSERT INTO user_behavior (name, url) VALUES (?,?)",
                        (statement, event) -> {
                            statement.setString(1, event.user);
                            statement.setString(2, event.url);
                        },
                        JdbcExecutionOptions
                                .builder()
                                .withBatchSize(1000)
                                .withBatchIntervalMs(200)
                                .withMaxRetries(5)
                                .build(),
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withUrl("jdbc:mysql://localhost:3306/atguigudb")
                                .withDriverName("com.mysql.cj.jdbc.Driver")
                                .withUsername("root")
                                .withPassword("123456")
                                .build()
                        )
        );

        env.execute();
    }
}
